package com.example.kafka.demo;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;

/**
 * @application:
 * @author: LiYuhang
 * @Date: 2021/6/26 15:33
 * @version: 0.1
 * @Description :
 * @Modified By:
 */
public class Demo2 {
    public static final String sourceTopic = "test";
    public static final String targetTopic = "targetTopic";
    public static final String groupId = "topic.group2";

    /**
     * topic流式增强 sourceTopic -> filter() -> targetTopic
     * @param args
     */
    public static void main(String[] args) {
        ConsumerDemo1 consumerDemo1 = new ConsumerDemo1(sourceTopic,groupId);
        com.example.kafka.demo.ProducerDemo1 producerDemo1 = new ProducerDemo1(sourceTopic);
        com.example.kafka.demo.ProducerDemo1 producerDemo2 = new ProducerDemo1(targetTopic);

        List<CompletableFuture> futureList = startJob(consumerDemo1,producerDemo1,producerDemo2);

        futureList.forEach(CompletableFuture::join);
    }


    private static List<CompletableFuture> startJob(ConsumerDemo1 consumerDemo1,ProducerDemo1 producerDemo1,ProducerDemo1 producerDemo2) {
        CompletableFuture future1 = CompletableFuture.runAsync(() -> {
            producerDemo1.run();
        });
        CompletableFuture future2 = CompletableFuture.runAsync(() -> {
            consumerDemo1.run(producerDemo2);
        });
        return Arrays.asList(future1,future2);
    }
}
